package org.hadoop.input.line;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
public class LineNumberInputFormat extends FileInputFormat<LongWritable, Text> {
    @Override
    public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {
        RecordReader<LongWritable, Text> reader = new LineTextReader();
        reader.initialize(split, context);
        return reader;
    }
    public static class LineTextReader extends RecordReader<LongWritable, Text> {
        private Long fileSize;//总大小，用于计算这个块的的读取比例
        private Long sum = 0L;//读取到多少字节了，可与fileSize计算出读取比例
        private BufferedReader br;//读取一次读取一行IO对象
        private LongWritable line = new LongWritable(0L);//行号
        private Text value = new Text();//行数据
        @Override
        public void initialize(InputSplit split, TaskAttemptContext context)
                throws IOException, InterruptedException {
            if (split instanceof FileSplit) {//判断是否是文件类型
                FileSplit fileSplit = (FileSplit) split;
                fileSize = fileSplit.getLength();
                Path path = fileSplit.getPath();
                FileSystem fs = FileSystem.get(context.getConfiguration());
                br = new BufferedReader(new InputStreamReader(fs.open(path)));
            }
        }
        /**
         * 用于返回是否还有下一行数据
         */
        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
            String line = br.readLine();
            if (line != null) {
                this.line.set(this.line.get() + 1);
                value.set(line);
                sum += line.getBytes().length;//读取到多少数据了
                return true;
            }
            return false;
        }
        /**
         * 获取当前key值,只有nextKeyValue返回true时才会调用这个方法
         */
        @Override
        public LongWritable getCurrentKey() throws IOException, InterruptedException {
            return this.line;
        }
        /**
         * 获取当前行数据
         */
        @Override
        public Text getCurrentValue() throws IOException, InterruptedException {
            return value;
        }
        /**
         * 获取进度
         */
        @Override
        public float getProgress() throws IOException, InterruptedException {
            return 1.0F * sum / fileSize;
        }
        /**
         * 最后关闭输入流
         */
        @Override
        public void close() throws IOException {
            if (br != null) {
                br.close();
                br = null;
            }
        }
    }
}